package com.hu.flink12.api

import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.configuration.Configuration
import org.apache.flink.metrics.Counter
import org.apache.flink.streaming.api.scala._

/**
 * @Author: hujianjun
 * @Create Date: 2021/2/2 12:57
 * @Describe: flink监控指标统计count值
 */
object FlinkCounterMetrics {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()
    env.setParallelism(1)
    val inputSource = env.socketTextStream("localhost", 7777)
    inputSource.map(new RichMapFunction[String, (String, Integer)] {
      private var wordCount: Counter = _

      override def open(parameters: Configuration): Unit = {
        wordCount = getRuntimeContext.getMetricGroup().addGroup("mapCount").counter("wordCount")
      }

      override def map(in: String): (String, Integer) = {
        wordCount.inc()
        new Tuple2[String, Integer](in, 1)
      }
    })

    env.execute()
  }
}
